Event Triggers

Event triggers allow BindAI applications to react when an event is published through an EventBus. The EventTrigger class connects an event source to an application-defined callable:

Event Source
     |
     v
  EventBus
     |
     v
EventTrigger
     |
     v
  Target

This provides a simple foundation for event-driven automation.

EventTrigger

EventTrigger is provided by the bindai-automation package:

from bindai_automation import EventTrigger

It requires three values:

bus — the EventBus that publishes events.
event_name — the event name to subscribe to.
target — the callable that receives matching events.

For example:

trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_event,
)

Creating a Target

The target is a normal callable that receives the published event.

def handle_event(event):
    print("Event received:", event.name)

The target can be any compatible callable, including a function or callable object. The automation package does not impose a particular target implementation.

Attaching a Trigger

Creating an EventTrigger does not automatically subscribe it to the event bus. Call:

trigger.attach()

to establish the subscription. A typical setup is:

from bindai_automation import EventTrigger


def handle_event(event):
    print("Received:", event.name)


trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_event,
)

trigger.attach()

After attachment, matching events are delivered to handle_event.

Detaching a Trigger

A trigger can be removed from the event bus with:

trigger.detach()

After detaching, the trigger no longer receives events from that subscription. For example:

trigger.attach()

# Trigger is active.

trigger.detach()

# Trigger is no longer subscribed.

Calling detach() when the trigger is already detached is safe.

Duplicate Attach Protection

EventTrigger prevents the same trigger from being attached multiple times. For example:

trigger.attach()
trigger.attach()

does not create two subscriptions for the same trigger. This is important for applications that initialize automation components more than once. The trigger internally tracks whether it is attached.

Enabled and Disabled State

An attached trigger can be temporarily disabled.

trigger.disable()

A disabled trigger remains subscribed to the event bus but does not invoke its target when an event arrives. It can be enabled again:

trigger.enable()

This gives applications two different ways to stop automation:

Disable

trigger.disable()

Keep the subscription but temporarily ignore events.

Detach

trigger.detach()

Remove the event subscription entirely.

Trigger Lifecycle

The typical lifecycle is:

Create
  |
  v
Attach
  |
  v
Enabled
  |
  +---- Disable
  |       |
  |       v
  |    Disabled
  |       |
  |       v
  |     Enable
  |       |
  +-------+
  |
  v
Detach

For example:

trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_event,
)

trigger.attach()

# Use automation...

trigger.disable()

# Temporarily disabled.

trigger.enable()

# Active again.

trigger.detach()

Event Names

An EventTrigger subscribes to a specific event name. For example:

trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_event,
)

The underlying EventBus uses the event’s name property to determine which handlers should receive it. The exact event names available depend on the events implemented by the BindAI event system.

BindAI Event Objects

BindAI events inherit from the core Event abstraction. An event contains common information such as:

id
timestamp
payload
name

For example, a target can inspect:

def handle_event(event):
    print(event.name)
    print(event.id)
    print(event.timestamp)
    print(event.payload)

The EventTrigger passes the event object to the target without transforming it.

Agent Event Sources

Agents expose an event bus through:

agent.events

This makes agent events a natural source for event-driven automation. For example:

from bindai_automation import EventTrigger


def handle_event(event):
    print("Agent event:", event.name)


trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_event,
)

trigger.attach()

The flow is:

Agent
  |
  v
agent.events
  |
  v
EventTrigger
  |
  v
handle_event()

Tool Execution Events

Agent tool execution publishes a ToolExecutedEvent through the agent’s event bus. A trigger can react to that event. For example:

def handle_tool_event(event):
    print("Tool executed:", event.payload)


trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_tool_event,
)

trigger.attach()

The resulting flow is:

Agent
  |
  v
execute_tool()
  |
  v
ToolExecutedEvent
  |
  v
EventBus
  |
  v
EventTrigger
  |
  v
handle_tool_event()

The exact event name should follow the EventTypes value used by the installed BindAI version.

Wildcard Events

The underlying EventBus supports wildcard subscriptions using:

However, EventTrigger is configured with one specific event_name and subscribes directly to that name. Applications that require broader event routing should use the underlying EventBus capabilities or a higher-level routing abstraction rather than assuming that EventTrigger itself provides pattern matching. Advanced event routing is outside the current EventTrigger abstraction.

Multiple Event Triggers

An application can create multiple triggers for different events.

tool_trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_tool_event,
)

finished_trigger = EventTrigger(
    bus=agent.events,
    event_name="agent.finished",
    target=handle_finished_event,
)

They can then be attached independently:

tool_trigger.attach()
finished_trigger.attach()

This keeps different automation rules separate.

Multiple Targets

Different triggers can use different targets:

tool_trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=notify_team,
)

finished_trigger = EventTrigger(
    bus=agent.events,
    event_name="agent.finished",
    target=record_completion,
)

This allows each event to have its own automation behavior.

Shared Event Sources

Multiple triggers can subscribe to the same event bus. For example:

trigger_a = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_tool,
)

trigger_b = EventTrigger(
    bus=agent.events,
    event_name="agent.finished",
    target=handle_finished,
)

Both use:

agent.events

but react to different event names. The event bus remains the shared event source.

Event Trigger and EventBus

The relationship between the two components is:

EventBus
   |
   +---- handler A
   |
   +---- EventTrigger
   |          |
   |          v
   |        target
   |
   +---- handler B

EventTrigger is therefore an adapter between the core event system and automation logic. It does not replace EventBus.

Event Handler Isolation

The core EventBus isolates exceptions raised by individual event handlers. For example, if a target raises an exception:

def handle_event(event):
    raise RuntimeError("Automation failed")

the event bus handles handler failures according to its own event-dispatch behavior rather than making EventTrigger responsible for retry or recovery. This behavior comes from the core EventBus, not from a separate retry system inside EventTrigger. Applications requiring specific failure handling should implement that behavior at the appropriate higher-level automation or execution layer.

Synchronous Event Dispatch

The current BindAI EventBus is synchronous. When an event is published, subscribed handlers are invoked during event publication. Conceptually:

publish(event)
     |
     v
EventTrigger
     |
     v
target(event)

EventTrigger itself does not create a background worker. It also does not persist events or queue automation jobs. For automation definitions that require background execution, BindAI provides AutomationWorker as a separate execution abstraction. This keeps event detection and background execution as separate concerns:

EventBus
   |
   v
EventTrigger
   |
   v
Target
   |
   v
AutomationWorker
   |
   v
AutomationRun

Background Automation Execution

AutomationWorker executes AutomationDefinition instances and manages their AutomationRun lifecycle. It is separate from EventTrigger. An application can use an event trigger to decide when an automation should start and then use an AutomationWorker to execute the automation in the background. For example:

from bindai_automation import AutomationWorker, EventTrigger


worker = AutomationWorker()


def handle_event(event):
    worker.submit(
        automation,
        input=event.payload,
    )


trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_event,
)

trigger.attach()

The resulting architecture is:

Event Source
    |
    v
 EventBus
    |
    v
EventTrigger
    |
    v
Target
    |
    v
AutomationWorker
    |
    v
AutomationRun
    |
    +----------------------+
    |                      |
    v                      v
AutomationStateStore  AutomationRunHistory

The trigger remains responsible for event subscription and lifecycle. The worker is responsible for automation execution and run lifecycle management. When the application shuts down, the worker should also be shut down:

trigger.detach()
worker.shutdown()

The worker can alternatively be used as a context manager:

with AutomationWorker() as worker:
    # Configure and run automations.
    ...

No Built-In Retry

The current EventTrigger does not implement retry policies. There is no built-in API such as:

trigger.retry(...)

If automation requires retry behavior, it should be implemented at a higher-level workflow or automation execution layer. This keeps the basic event-trigger abstraction small and predictable.

No Trigger-Level Persistent State

The current EventTrigger does not persist:

Events
Trigger executions
Trigger history
Retry state

Its own state is limited to the trigger configuration and whether the trigger is enabled and attached. The broader automation package provides separate abstractions for automation execution state and run history:

from bindai_automation import (
    AutomationRunHistory,
    AutomationStateStore,
    MemoryAutomationRunHistory,
    MemoryAutomationStateStore,
)

These persistence abstractions belong to the automation execution layer rather than to EventTrigger itself. In other words:

EventTrigger
    |
    +-- subscription lifecycle
    +-- enabled/disabled state
    |
    v
Automation execution
    |
    +-- AutomationRun
    +-- AutomationStateStore
    +-- AutomationRunHistory

The trigger does not automatically create or persist an AutomationRun.

Using a Function as a Target

A simple function is often sufficient:

def notify(event):
    print("Notification:", event.name)


trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=notify,
)

trigger.attach()

This is useful for small application-level automation rules.

Using a Callable Object

A callable class can also be used:

class EventHandler:
    def __call__(self, event):
        print("Handling:", event.name)


handler = EventHandler()

trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handler,
)

trigger.attach()

The target only needs to satisfy the callable interface expected by EventTrigger.

Calling a Workflow from a Target

An application can use a target to invoke other BindAI functionality. Conceptually:

def start_workflow(event):
    workflow = create_workflow(event)

    # Invoke the workflow using the application's
    # supported execution mechanism.

Then:

trigger = EventTrigger(
    bus=agent.events,
    event_name="agent.finished",
    target=start_workflow,
)

The trigger itself does not define the workflow execution API. The target is responsible for invoking the appropriate application behavior.

Calling an Agent from a Target

The same pattern can be used with an agent:

def process_event(event):
    agent.run(f"Process event: {event.payload}")

Then:

trigger = EventTrigger(
    bus=source.events,
    event_name="tool.executed",
    target=process_event,
)

Applications should validate event data before passing it into prompts or other agent inputs.

Using Connections from a Target

A target can also interact with a BindAI connection. For example:

def notify_external_service(event):
    connection.send(
        {
            "event": event.name,
            "payload": event.payload,
        }
    )

The resulting architecture is:

Event
  |
  v
EventBus
  |
  v
EventTrigger
  |
  v
Target
  |
  v
Connection
  |
  v
External Service

The trigger remains responsible only for connecting the event to the target.

TriggerRegistry

Multiple triggers can be managed through TriggerRegistry.

from bindai_automation import TriggerRegistry

registry = TriggerRegistry()

registry.register(
    "tool-events",
    trigger,
)

Retrieve it:

trigger = registry.get("tool-events")

Or use a non-throwing lookup:

trigger = registry.get_or_none("tool-events")

Managing Registered Triggers

The registry supports:

registry.contains("tool-events")

registry.remove("tool-events")

registry.clear()

It can also enumerate registered triggers:

registry.keys()

registry.values()

registry.items()

And supports:

len(registry)

The registry is useful when an application needs to manage trigger instances centrally.

Registry Does Not Attach Triggers

Registering a trigger does not itself attach it to an event bus. For example:

registry.register(
    "tool-events",
    trigger,
)

does not replace:

trigger.attach()

Trigger registration and trigger lifecycle remain separate concerns.

Security

Event-driven automation can cause actions without direct user interaction. Targets should therefore be treated as potentially privileged application logic. Recommended practices include:

Validate event payloads.
Avoid exposing secrets through events.
Restrict state-changing actions.
Apply appropriate authorization.
Keep credentials outside event payloads.
Avoid passing untrusted event data directly into sensitive operations.
Audit important automated actions.

For example, an event should not automatically become permission to perform an unrelated privileged operation.

Event Payloads

Event payloads can contain application-specific information. A target can inspect:

def handle_event(event):
    payload = event.payload

Applications should keep payloads limited to the information required by the automation rule. Avoid placing:

API keys
Access tokens
Passwords
Private credentials

inside event payloads unless there is an explicit and secure reason.

Testing Event Triggers

Event triggers can be tested without external services. A typical test can:

Create an event bus.
Create a test target.
Create an event trigger.
Attach the trigger.
Publish an event.
Verify that the target was called.
Detach the trigger.
Verify that later events are ignored.

Conceptually:

events = EventBus()

received = []


def handle(event):
    received.append(event)


trigger = EventTrigger(
    bus=events,
    event_name="test.event",
    target=handle,
)

trigger.attach()

The test can then publish an appropriate event through the bus and inspect received.

Testing Disable Behavior

Disable behavior should also be tested.

trigger.attach()
trigger.disable()

An event published while the trigger is disabled should not invoke the target. After:

trigger.enable()

matching events should again invoke the target.

Testing Detach Behavior

Detach behavior should verify that the subscription is removed:

trigger.attach()
trigger.detach()

Events published afterward should not reach the trigger. This is particularly important for applications that dynamically create and remove automation rules.

Testing Duplicate Attachment

Applications can verify that:

trigger.attach()
trigger.attach()

does not cause duplicate target invocations. The trigger tracks its attached state and avoids creating duplicate subscriptions.

Recommended Lifecycle

For application startup:

trigger = EventTrigger(
    bus=agent.events,
    event_name="tool.executed",
    target=handle_event,
)

trigger.attach()

During runtime:

# Trigger remains active.

When temporarily disabling automation:

trigger.disable()

When re-enabling:

trigger.enable()

When shutting down:

trigger.detach()

This makes the trigger lifecycle explicit. If a background AutomationWorker is being used, it should also be shut down during application shutdown:

trigger.detach()
worker.shutdown()

Event Triggers and Automation Architecture

Event triggers form the event-driven entry layer of the broader BindAI Automation architecture. The trigger layer is:

             BindAI Event
                  |
                  v
               EventBus
                  |
                  v
             EventTrigger
                  |
                  v
                Target

The target can then invoke a higher-level automation execution path:

Event
  |
  v
Trigger
  |
  v
Automation Target
  |
  v
Automation Definition
  |
  v
Automation Run
  |
  +---- Execution State
  |
  +---- Run History

The current automation package provides foundations for:

AutomationDefinition
AutomationRun
AutomationStateStore
MemoryAutomationStateStore
AutomationRunHistory
MemoryAutomationRunHistory
AutomationWorker

These components are separate from EventTrigger. For example, an application may use an event trigger to initiate an automation and then use the automation execution layer to track the resulting run. The trigger itself does not automatically perform these operations.

Best Practices

Use event triggers when:

An application needs to react to BindAI events.
Event handling should remain separate from the event source.
A small callable is sufficient for the automation action.
Trigger lifecycle needs to be controlled explicitly.
Multiple event-driven actions need to coexist.

Prefer to:

Keep targets focused.
Validate event payloads.
Attach triggers deliberately.
Detach triggers when no longer needed.
Disable triggers when temporary suspension is sufficient.
Keep secrets outside event payloads.
Test trigger lifecycle independently.
Keep execution state and run history in the appropriate automation layer.
Use AutomationWorker when automation definitions require background execution.

Avoid:

Putting large orchestration systems directly inside a target.
Treating an event as automatic authorization for privileged operations.
Assuming that event dispatch is asynchronous.
Assuming built-in retries.
Assuming that EventTrigger itself provides persistent state or run history.
Assuming that EventTrigger itself manages background worker lifecycle.

Current Scope

The current EventTrigger implementation provides:

Event bus subscription
Named event matching through EventBus
Callable targets
Attach and detach lifecycle
Enable and disable state
Duplicate attachment protection

The broader automation package additionally provides foundational execution-state, run-history, and background-worker abstractions, but those capabilities are not automatically performed by EventTrigger. AutomationWorker provides explicit background execution for AutomationDefinition instances. Advanced event routing, retry policies, persistent storage backends, and other higher-level execution capabilities remain separate concerns.

Summary

EventTrigger provides a small adapter between BindAI’s event system and application automation logic. The core flow is:

Event Source
     |
     v
  EventBus
     |
     v
EventTrigger
     |
     v
  Target

A trigger can be:

trigger.attach()

temporarily disabled:

trigger.disable()

re-enabled:

trigger.enable()

and eventually detached:

trigger.detach()

The current implementation deliberately keeps event triggering simple. The broader automation package provides separate abstractions for automation definitions, execution state, run history, and background execution. An application can therefore compose event-driven triggering with background automation execution:

Event
  |
  v
EventBus
  |
  v
EventTrigger
  |
  v
Target
  |
  v
AutomationWorker
  |
  v
AutomationRun
  |
  +---- AutomationStateStore
  |
  +---- AutomationRunHistory

This separation keeps EventTrigger focused on event subscription and lifecycle while allowing the broader automation layer to handle execution, state, history, and background work.

​Event Triggers

​EventTrigger

​Creating a Target

​Attaching a Trigger

​Detaching a Trigger

​Duplicate Attach Protection

​Enabled and Disabled State

​Disable

​Detach

​Trigger Lifecycle

​Event Names

​BindAI Event Objects

​Agent Event Sources

​Tool Execution Events

​Wildcard Events

​Multiple Event Triggers

​Multiple Targets

​Shared Event Sources

​Event Trigger and EventBus

​Event Handler Isolation

​Synchronous Event Dispatch

​Background Automation Execution

​No Built-In Retry

​No Trigger-Level Persistent State

​Using a Function as a Target

​Using a Callable Object

​Calling a Workflow from a Target

​Calling an Agent from a Target

​Using Connections from a Target

​TriggerRegistry

​Managing Registered Triggers

​Registry Does Not Attach Triggers

​Security

​Event Payloads

​Testing Event Triggers

​Testing Disable Behavior

​Testing Detach Behavior

​Testing Duplicate Attachment

​Recommended Lifecycle

​Event Triggers and Automation Architecture

​Best Practices

​Current Scope

​Summary